package com.wzl.producer.controller.kafkaController.basicsProducer;

import com.alibaba.fastjson.JSONObject;
import com.wzl.common.VkPublicResponse;
import com.wzl.common.constant.VkPublicConstants;
import com.wzl.common.errcode.VkErrorCodeEnum;
import com.wzl.common.exception.VkPublicException;
import com.wzl.common.pojo.VkSmbmsUser;
import com.wzl.producer.config.kafkaConfig.KafkaConfig;
import com.wzl.producer.service.kafkaService.basicsProducer.MqBasicsKafkaProService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Controller;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.ResponseBody;

import javax.annotation.Resource;

/**
 * kafka消息队列基本使用
 *
 * https://blog.csdn.net/yuanlong122716/article/details/105160545?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522166304793016800186521704%2522%252C%2522scm%2522%253A%252220140713.130102334..%2522%257D&request_id=166304793016800186521704&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~all~top_positive~default-1-105160545-null-null.142^v47^pc_rank_34_default_3,201^v3^control_2&utm_term=springboot%E9%9B%86%E6%88%90kafka&spm=1018.2226.3001.4187
 *
 * @author weizhongli
 * @since 2022-09-13 10:59
 */
@Controller
@Slf4j
@ResponseBody
public class MqBasicsKafkaProController {

    @Resource
    private MqBasicsKafkaProService mqBasicsKafkaProService;

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 简单发送消息
     */
    @GetMapping("/kafka/normal/{id}")
    public VkPublicResponse sendMessage1(@PathVariable("id") int id) {
        VkSmbmsUser user = new VkSmbmsUser();
        try {
            user =  mqBasicsKafkaProService.queryUserById(id);
            kafkaTemplate.send(KafkaConfig.TOPIC, JSONObject.toJSONString(user));
        }catch (Exception e) {
            e.printStackTrace();
            return new VkPublicResponse(VkErrorCodeEnum.MESSAGE_SEND_ERROR,"消费信息失败，messageId:"+user.getId()+" 原因:"+e.getMessage());
        }
        return new VkPublicResponse(VkPublicConstants.SUCCESS_CODE, VkPublicConstants.SUCCESS_MESSAGE,"信息投递成功，messageId:"+user.getId());
    }

    /**
     * 生产者加入回调
     * 在回调方法中监控消息是否发送成功
     */
    @GetMapping("/kafka/callbackOne/{id}")
    public VkPublicResponse sendMessage2(@PathVariable("id") int id) {
        VkSmbmsUser user = new VkSmbmsUser();
        try {
            user =  mqBasicsKafkaProService.queryUserById(id);
            kafkaTemplate.send(KafkaConfig.TOPIC, JSONObject.toJSONString(user)).addCallback(success -> {
                // 消息发送到的topic
                String topic = success.getRecordMetadata().topic();
                // 消息发送到的分区
                int partition = success.getRecordMetadata().partition();
                // 消息在分区内的offset
                long offset = success.getRecordMetadata().offset();
                log.info("消息发送到的Topic:" + topic + ",消息发送到的分区:" + partition + ",消息在分区内的offset:" + offset);
            }, fail ->{
               log.error("发送消息失败:" + fail.getMessage());
            });
        }catch (Exception e) {
            e.printStackTrace();
            return new VkPublicResponse(VkErrorCodeEnum.MESSAGE_SEND_ERROR,"消费信息失败，messageId:"+ user.getId()+" 原因:"+e.getMessage());
        }
        return new VkPublicResponse(VkPublicConstants.SUCCESS_CODE, VkPublicConstants.SUCCESS_MESSAGE,"信息投递成功:" + user);
    }

    /**
     * 生产者加入回调
     * 失败时做补偿处理
     */
    @GetMapping("/kafka/callbackTwo/{id}")
    public VkPublicResponse sendMessage3(@PathVariable("id") int id) {
        VkSmbmsUser user = new VkSmbmsUser();
        try {
            user =  mqBasicsKafkaProService.queryUserById(id);
            kafkaTemplate.send(KafkaConfig.TOPIC1, JSONObject.toJSONString(user)).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                 @Override
                 public void onFailure(Throwable throwable) {
                     log.error("发送消息失败：" + throwable.getMessage());
                 }
                 @Override
                 public void onSuccess(SendResult<String, String> stringStringSendResult) {
                     String topic = stringStringSendResult.getRecordMetadata().topic();
                     // 消息发送到的分区
                     int partition = stringStringSendResult.getRecordMetadata().partition();
                     // 消息在分区内的offset
                     long offset = stringStringSendResult.getRecordMetadata().offset();
                     log.info("消息发送到的Topic:" + topic + ",消息发送到的分区:" + partition + ",消息在分区内的offset:" + offset);
                 }
             }
         );
        }catch (Exception e) {
            e.printStackTrace();
            return new VkPublicResponse(VkErrorCodeEnum.MESSAGE_SEND_ERROR,"消费信息失败，messageId:"+user.getId()+" 原因:"+e.getMessage());
        }
        return new VkPublicResponse(VkPublicConstants.SUCCESS_CODE, VkPublicConstants.SUCCESS_MESSAGE,"信息投递成功:" + user);
    }

    /**
     * kafka事务提交
     * 如果在发送消息时需要创建事务，可以使用 KafkaTemplate 的 executeInTransaction 方法来声明事务，
     */
    @GetMapping("/kafka/transaction")
    public VkPublicResponse sendMessage7(){
        // 声明事务：后面报错消息不会发出去
        kafkaTemplate.executeInTransaction(operations -> {
            operations.send(KafkaConfig.TOPIC1,"事务测试");
            try {
                int a = 1 / 0;
                throw new VkPublicException(VkErrorCodeEnum.MESSAGE_SEND_ERROR);
            } catch (VkPublicException e) {
                throw new RuntimeException(e);
            }
        });

//        // 不声明事务：后面报错但前面消息已经发送成功了
//        kafkaTemplate.send(KafkaConfig.TOPIC,"事务测试");
//        throw new VkPublicException(VkErrorCodeEnum.MESSAGE_SEND_ERROR);
        return new VkPublicResponse();
    }

    /**
     * 生产者事务发送：需配置transaction-id-prefix开启事务
     * @return
     */
    @Transactional
    @GetMapping("/kafka/transaction1")
    public VkPublicResponse sendMessage8(){
        kafkaTemplate.send(KafkaConfig.TOPIC1,"事务测试1111");
        int a = 1 / 0;
        kafkaTemplate.send(KafkaConfig.TOPIC1, "_____" + "事务测试2222");
        return new VkPublicResponse();
    }


}